package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.util.Random

object Demo15Pi {
  def main(args: Array[String]): Unit = {
    // 生成 [-1,1]内的随机数

    // Random.nextDouble() 默认生成0.0 -> 1.0之间的小数

    // slices表示生成多少个任务，cnt表示每个任务内生成多少个点
    val cnt = 10000000
    val slices = 100
    // local[*] 表示将并行度设置为跟电脑的cpu线程数一致
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Demo15Pi")
    val sc: SparkContext = new SparkContext(conf)

    // seqRDD的分区数等于slices
    // parallelize 支持在传入一个参数numSlices 默认是同并行度一致，可以手动指定 表示最后生成的RDD分区数是多少
    // 最终会决定task的数量
    val seqRDD: RDD[Int] = sc.parallelize(0 to cnt * slices, slices)

    // 随机生成N个[-1,1]内的点
    val pointRDD: RDD[(Double, Double)] = seqRDD.map(i => {
      val x: Double = Random.nextDouble() * 2 - 1
      val y: Double = Random.nextDouble() * 2 - 1
      (x, y)
    })

    // 过滤出圆内的点
    val circlePointNum: Long = pointRDD.filter {
      case (x: Double, y: Double) =>
        val res: Double = x * x + y * y
        res <= 1
    }.count()

    val pi: Double = circlePointNum.toDouble / cnt / slices * 4
    println("PI is " + pi)

  }
}
